package mylab.starters.mq.impl.rabbitmq;


import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import mylab.starters.mq.api.Message;
import mylab.starters.mq.api.Properties;
import mylab.starters.mq.config.MqConst;
import mylab.starters.mq.impl.AbstractProducer;
import mylab.utils.spring.SpringUtil;


public class RabbitProducer extends AbstractProducer {

    ConnectionFactory factory;
    Connection connection;

    public RabbitProducer(Properties properties, String groupId, String clientId) {
        super(properties, groupId, clientId);
        factory = SpringUtil.getBean(ConnectionFactory.class);
    }

    @Override
    protected void onStart() throws Exception {
        connection = factory.newConnection();
    }

    @Override
    protected void onStop() throws Exception {
        connection.close();
    }

    @Override
    protected Object onSend(Message message) throws Exception {
        assert message != null;
        assert message.getTopic() != null;
        assert message.getContent() != null;

        String exchangeName = MqConst.RABBIT_EXCHANGE;
        String queueName = message.getTopic();
        Channel channel = connection.createChannel();
        channel.exchangeDeclare(exchangeName, BuiltinExchangeType.DIRECT);
        channel.queueDeclare(queueName, false, false, false, null);
        channel.basicPublish(exchangeName, queueName, null, message.getContent().getBytes());
        channel.close();

        return null;
    }

}
